# -*- coding: utf-8 -*-
"""
通用数据同步质量报表 Provider，支持多源表、单目标表、可选筛选、可选统计字段。
"""
from __future__ import annotations

import json
import logging
from datetime import datetime, timedelta
from typing import Any, Dict, List, Optional

from app.blueprints.reports.providers.base import BaseProvider, SafeDict
from app.services.report_service import ReportTemplateService
from app.utils.logger import FlinkLogger

# 延迟初始化 logger
logger = None


def get_logger():
    global logger
    if logger is None:
        logger = FlinkLogger.get_logger(__name__)
    return logger


def escape_column_name(column_name: str) -> str:
    """
    转义列名，使用反引号包裹以支持中文和特殊字符
    """
    if not column_name:
        return column_name
    # 如果已经用反引号包裹，直接返回
    if column_name.startswith("`") and column_name.endswith("`"):
        return column_name
    return f"`{column_name}`"


BASE_DISPLAY_FIELD_ORDER = [
    "source_table",
    "target_table",
    "source_total_count",
    "target_total_count",
    "sync_rate",
    "data_status",
]


class GenericDataQualityProvider(BaseProvider):
    """通用数据同步质量报表提供者"""

    def _normalize_and_sort_display_fields(
        self, display_fields: List[str], field_labels: Dict[str, str]
    ) -> List[str]:
        """将显示字段转换为英文字段名并按预设顺序排序"""
        if not display_fields:
            return []

        label_to_field = {
            label: key for key, label in field_labels.items() if isinstance(label, str)
        }
        normalized: List[str] = []

        for field in display_fields:
            canonical = label_to_field.get(field, field)
            if canonical not in normalized:
                normalized.append(canonical)

        sorted_fields: List[str] = []
        for field in BASE_DISPLAY_FIELD_ORDER:
            if field in normalized:
                sorted_fields.append(field)

        for field in normalized:
            if field not in BASE_DISPLAY_FIELD_ORDER:
                sorted_fields.append(field)

        return sorted_fields

    def preview_sql(self) -> str:
        """预览SQL（不执行）"""
        return self._build_sql()

    def _build_time_condition(
        self, time_start: Optional[str], time_end: Optional[str], time_field: str
    ) -> str:
        """构建时间筛选条件"""
        conditions = []
        if time_start and time_field:
            conditions.append(f"{time_field} >= '{time_start}'")
        if time_end and time_field:
            conditions.append(f"{time_field} <= '{time_end}'")
        return " AND " + " AND ".join(conditions) if conditions else ""

    def _build_source_union_sql(
        self,
        source_tables: Dict[str, Any],
        join_field: str,
        time_start: Optional[str],
        time_end: Optional[str],
        stat_field_mappings: Optional[Dict[str, Any]] = None,
    ) -> tuple[str, str, List[str]]:
        """
        构建源表 UNION SQL
        返回: (source_union_sql, source_ids_sql, source_table_list)
        """
        source_union_parts = []
        source_ids_parts = []
        source_table_list = []

        if isinstance(source_tables, dict):
            data_source_id = source_tables.get("data_source_id")
            databases = source_tables.get("databases", [])
            database_name = source_tables.get("database_name")
            table_names = source_tables.get("table_names", [])
            tables_config = source_tables.get("tables", [])

            if databases:
                # 多数据库配置
                for db_config in databases:
                    db_name = db_config.get("database_name")
                    tbl_names = db_config.get("table_names", [])
                    tbls_config = db_config.get("tables", [])
                    
                    # 从每个数据库配置中获取过滤条件和统计字段
                    db_filter_conditions = db_config.get("filter_conditions", [])
                    db_stat_fields = db_config.get("stat_fields", {})
                    
                    # 解析统计字段配置
                    if isinstance(db_stat_fields, str):
                        try:
                            db_stat_fields = json.loads(db_stat_fields) if db_stat_fields.strip() else {}
                        except Exception:
                            db_stat_fields = {}
                    elif not isinstance(db_stat_fields, dict):
                        db_stat_fields = {}

                    if not db_name or not tbl_names:
                        continue

                    for tbl_name in tbl_names:
                        tbl_config = next(
                            (t for t in tbls_config if t.get("table_name") == tbl_name),
                            {},
                        )
                        # 从表配置中读取过滤条件和统计字段（表级别）
                        tbl_filter_conditions = tbl_config.get("filter_conditions", [])
                        tbl_stat_fields = tbl_config.get("stat_fields", {})
                        if isinstance(tbl_stat_fields, str):
                            try:
                                tbl_stat_fields = json.loads(tbl_stat_fields) if tbl_stat_fields.strip() else {}
                            except Exception:
                                tbl_stat_fields = {}
                        elif not isinstance(tbl_stat_fields, dict):
                            tbl_stat_fields = {}
                        
                        # 如果表配置中没有，则使用数据库级别的配置（向后兼容）
                        if not tbl_filter_conditions:
                            tbl_filter_conditions = db_filter_conditions
                        if not tbl_stat_fields:
                            tbl_stat_fields = db_stat_fields
                        
                        self._add_table_sql(
                            db_name,
                            tbl_name,
                            tbl_config,
                            join_field,
                            time_start,
                            time_end,
                            source_union_parts,
                            source_ids_parts,
                            filter_conditions=tbl_filter_conditions,
                            stat_fields=tbl_stat_fields,
                            stat_field_mappings=stat_field_mappings,
                        )
                        source_table_list.append(f"{db_name}.{tbl_name}")
            elif database_name and table_names:
                # 单数据库多表配置（兼容旧格式，从顶层获取过滤条件和统计字段）
                filter_conditions = source_tables.get("filter_conditions", [])
                stat_fields = source_tables.get("stat_fields", {})
                if isinstance(stat_fields, str):
                    try:
                        stat_fields = json.loads(stat_fields) if stat_fields.strip() else {}
                    except Exception:
                        stat_fields = {}
                elif not isinstance(stat_fields, dict):
                    stat_fields = {}
                
                for tbl_name in table_names:
                    tbl_config = next(
                        (t for t in tables_config if t.get("table_name") == tbl_name),
                        {},
                    )
                    self._add_table_sql(
                        database_name,
                        tbl_name,
                        tbl_config,
                        join_field,
                        time_start,
                        time_end,
                        source_union_parts,
                        source_ids_parts,
                        filter_conditions=filter_conditions,
                        stat_fields=stat_fields,
                        stat_field_mappings=stat_field_mappings,
                    )
                    source_table_list.append(f"{database_name}.{tbl_name}")
        elif isinstance(source_tables, list):
            # 列表格式配置
            for source_table in source_tables:
                db_name = source_table.get("database_name")
                tbl_name = source_table.get("table_name")
                if not all([db_name, tbl_name]):
                    continue
                self._add_table_sql(
                    db_name,
                    tbl_name,
                    source_table,
                    join_field,
                    time_start,
                    time_end,
                    source_union_parts,
                    source_ids_parts,
                    filter_conditions=filter_conditions,
                    stat_fields=stat_fields,
                    stat_field_mappings=stat_field_mappings,
                )
                source_table_list.append(f"{db_name}.{tbl_name}")
        else:
            raise ValueError("源表配置格式错误：应为对象或数组")

        if not source_union_parts:
            raise ValueError("没有有效的源表配置")

        source_union_sql = " UNION ALL ".join(source_union_parts)
        source_ids_sql = " UNION ALL ".join(source_ids_parts)

        return source_union_sql, source_ids_sql, source_table_list

    def _add_table_sql(
        self,
        database_name: str,
        table_name: str,
        table_config: Dict[str, Any],
        join_field: str,
        time_start: Optional[str],
        time_end: Optional[str],
        source_union_parts: List[str],
        source_ids_parts: List[str],
        filter_conditions: Optional[List[Dict[str, Any]]] = None,
        stat_fields: Optional[Dict[str, Any]] = None,
        stat_field_mappings: Optional[Dict[str, Any]] = None,
    ):
        """添加单个表的SQL"""
        use_distinct = table_config.get("use_distinct", False)
        # 如果未配置去重字段或为空字符串，则回退到关联字段
        distinct_field = table_config.get("distinct_field") or join_field
        time_field = table_config.get("time_field", "etl_date")

        table_ref = f"`{database_name}`.`{table_name}`"
        time_condition = self._build_time_condition(time_start, time_end, time_field)
        
        # 构建筛选条件
        filter_condition_sql = ""
        if filter_conditions:
            filter_condition_sql = self._build_filter_conditions(filter_conditions)
        
        combined_conditions = ""
        if time_condition:
            combined_conditions += time_condition
        if filter_condition_sql:
            combined_conditions += filter_condition_sql
        
        where_clause = f" WHERE 1=1{combined_conditions}" if combined_conditions else ""

        # 构建统计字段
        stat_select_clause = ""
        if stat_fields:
            stat_select, _ = self._build_stat_fields_sql(
                stat_fields, stat_field_mappings or {}, prefix="source_"
            )
            if stat_select:
                # 移除开头的逗号和空格（如果有）
                stat_select = stat_select.strip()
                if stat_select:
                    stat_select_clause = ", " + stat_select

        if use_distinct:
            select_clause = f"SELECT DISTINCT {distinct_field} AS {join_field}{stat_select_clause}"
        else:
            select_clause = f"SELECT {distinct_field} AS {join_field}{stat_select_clause}"

        sql_part = f"{select_clause} FROM {table_ref}{where_clause}"
        source_union_parts.append(sql_part)
        source_ids_parts.append(sql_part)

    def _build_filter_conditions(
        self, filter_conditions: List[Dict[str, Any]]
    ) -> str:
        """构建筛选条件SQL"""
        if not filter_conditions:
            return ""

        conditions = []
        for filter_cond in filter_conditions:
            field = filter_cond.get("field")
            operator = filter_cond.get("operator", "=")
            value = filter_cond.get("value")

            if not field:
                continue

            if operator.upper() == "IN":
                if isinstance(value, list):
                    values_str = ", ".join([f"'{v}'" for v in value])
                    conditions.append(f"{field} {operator} ({values_str})")
                else:
                    conditions.append(f"{field} {operator} ({value})")
            elif operator.upper() in ("LIKE", "NOT LIKE"):
                conditions.append(f"{field} {operator} '{value}'")
            else:
                conditions.append(f"{field} {operator} '{value}'")

        return " AND " + " AND ".join(conditions) if conditions else ""

    def _build_stat_fields_sql(
        self, stat_fields: Dict[str, Any], stat_field_mappings: Dict[str, Any], prefix: str = "target_"
    ) -> tuple[str, List[str]]:
        """
        构建统计字段SQL
        返回: (stat_select_clause, stat_output_fields)
        """
        # 根据前缀决定是否包含总数统计
        stat_selects = []
        stat_output_fields = []
        
        if prefix == "target_":
            stat_selects.append(f"COUNT(*) AS {escape_column_name('target_total_count')}")
            stat_output_fields.append("target_total_count")

        if not stat_fields:
            return ", ".join(stat_selects) if stat_selects else "", stat_output_fields

        for field_name, field_config in stat_fields.items():
            if not isinstance(field_config, dict):
                continue

            stat_type = field_config.get("type", "count")
            condition = field_config.get("condition", "")
            output_fields = field_config.get("output_fields", [])

            # 检查是否有字段映射配置
            field_mapping = stat_field_mappings.get(field_name, {})
            if field_mapping:
                output_fields = field_mapping.get("output_fields", output_fields)
                conditions = field_mapping.get(
                    "conditions", [condition] if condition else []
                )

                for idx, output_field in enumerate(output_fields):
                    escaped_output_field = escape_column_name(output_field)
                    if idx < len(conditions):
                        mapped_condition = conditions[idx]
                        stat_selects.append(
                            f"SUM(CASE WHEN {field_name} {mapped_condition} THEN 1 ELSE 0 END) AS {escaped_output_field}"
                        )
                    elif condition:
                        stat_selects.append(
                            f"SUM(CASE WHEN {field_name} {condition} THEN 1 ELSE 0 END) AS {escaped_output_field}"
                        )
                    else:
                        stat_selects.append(f"SUM({field_name}) AS {escaped_output_field}")
                    stat_output_fields.append(output_field)
                continue

            # 使用配置的输出字段
            if output_fields:
                for output_field in output_fields:
                    escaped_output_field = escape_column_name(output_field)
                    if condition:
                        stat_selects.append(
                            f"SUM(CASE WHEN {field_name} {condition} THEN 1 ELSE 0 END) AS {escaped_output_field}"
                        )
                    else:
                        stat_selects.append(f"SUM({field_name}) AS {escaped_output_field}")
                    stat_output_fields.append(output_field)
                continue

            # 默认生成字段名（使用前缀）
            if stat_type == "sum":
                if condition:
                    output_field = f"{prefix}{field_name}_count"
                    escaped_output_field = escape_column_name(output_field)
                    stat_selects.append(
                        f"SUM(CASE WHEN {field_name} {condition} THEN 1 ELSE 0 END) AS {escaped_output_field}"
                    )
                else:
                    output_field = f"{prefix}{field_name}_sum"
                    escaped_output_field = escape_column_name(output_field)
                    stat_selects.append(f"SUM({field_name}) AS {escaped_output_field}")
                stat_output_fields.append(output_field)
            elif stat_type == "count":
                if condition:
                    output_field = f"{prefix}{field_name}_count"
                    escaped_output_field = escape_column_name(output_field)
                    stat_selects.append(
                        f"COUNT(CASE WHEN {field_name} {condition} THEN 1 END) AS {escaped_output_field}"
                    )
                else:
                    output_field = f"{prefix}{field_name}_count"
                    escaped_output_field = escape_column_name(output_field)
                    stat_selects.append(f"COUNT({field_name}) AS {escaped_output_field}")
                stat_output_fields.append(output_field)
            elif stat_type == "avg":
                output_field = f"{prefix}{field_name}_avg"
                escaped_output_field = escape_column_name(output_field)
                stat_selects.append(f"AVG({field_name}) AS {escaped_output_field}")
                stat_output_fields.append(output_field)
            elif stat_type == "max":
                output_field = f"{prefix}{field_name}_max"
                escaped_output_field = escape_column_name(output_field)
                stat_selects.append(f"MAX({field_name}) AS {escaped_output_field}")
                stat_output_fields.append(output_field)
            elif stat_type == "min":
                output_field = f"{prefix}{field_name}_min"
                escaped_output_field = escape_column_name(output_field)
                stat_selects.append(f"MIN({field_name}) AS {escaped_output_field}")
                stat_output_fields.append(output_field)

        return ", ".join(stat_selects), stat_output_fields

    def _get_sql_template(self) -> Optional[str]:
        """从数据库读取SQL模板"""
        if not self.template_code:
            return None

        try:
            tpl = ReportTemplateService.get_template_by_code(self.template_code)
            if not tpl:
                return None

            default_params = tpl.get("default_params")
            if isinstance(default_params, str):
                try:
                    default_params = json.loads(default_params)
                except Exception:
                    return None

            if isinstance(default_params, dict):
                return default_params.get("sql_template")
        except Exception as exc:
            get_logger().warning(f"读取SQL模板失败: {exc}")

        return None

    def _get_field_labels(self) -> Dict[str, str]:
        """获取字段标签"""
        labels = {}

        # 从模板配置读取
        try:
            if self.template_code:
                tpl = ReportTemplateService.get_template_by_code(self.template_code)
                if tpl:
                    default_params = tpl.get("default_params")
                    if isinstance(default_params, str):
                        try:
                            default_params = json.loads(default_params)
                        except Exception:
                            default_params = {}
                    if isinstance(default_params, dict):
                        labels.update(default_params.get("field_labels", {}))
        except Exception:
            pass

        # 从运行时参数读取
        if isinstance(self.params, dict):
            runtime_labels = self.params.get("field_labels", {})
            if isinstance(runtime_labels, dict):
                labels.update(runtime_labels)

        return labels

    def _build_sql(self) -> str:
        """构建SQL"""
        source_tables = self.params.get("source_tables", {})
        target_config = self.params.get("target_table_config", {})
        filter_rules = self.params.get("filter_rules", {})

        if not source_tables:
            raise ValueError("源表配置不能为空")
        if not target_config:
            raise ValueError("目标表配置不能为空")

        # 时间范围处理
        time_range_type = source_tables.get("time_range_type", "all")
        time_start = source_tables.get("time_start")
        time_end = source_tables.get("time_end")

        if time_range_type == "yesterday":
            # 昨天：总是动态计算，确保定时任务自动更新
            now = datetime.now()
            start = datetime(now.year, now.month, now.day - 1)
            end = datetime(now.year, now.month, now.day)
            time_start = start.strftime("%Y-%m-%d %H:%M:%S")
            time_end = end.strftime("%Y-%m-%d %H:%M:%S")
        elif time_range_type in ("last7days", "recent"):
            # 最近7天：总是动态计算，确保定时任务自动更新
            now = datetime.now()
            start = datetime(now.year, now.month, now.day - 7)
            end = datetime(now.year, now.month, now.day)
            time_start = start.strftime("%Y-%m-%d %H:%M:%S")
            time_end = end.strftime("%Y-%m-%d %H:%M:%S")
        elif time_range_type == "thisMonth":
            # 本月：总是动态计算
            now = datetime.now()
            start = datetime(now.year, now.month, 1)
            end = datetime(now.year, now.month + 1, 1) - timedelta(seconds=1)
            time_start = start.strftime("%Y-%m-%d %H:%M:%S")
            time_end = end.strftime("%Y-%m-%d %H:%M:%S")
        elif time_range_type == "last_month":
            # 上月：总是动态计算，确保定时任务自动更新
            now = datetime.now()
            start = datetime(now.year, now.month - 1, 1)
            end = datetime(now.year, now.month, 1) - timedelta(seconds=1)
            time_start = start.strftime("%Y-%m-%d %H:%M:%S")
            time_end = end.strftime("%Y-%m-%d %H:%M:%S")
        elif time_range_type == "custom":
            if time_start == "":
                time_start = None
            if time_end == "":
                time_end = None
        else:
            time_start = None
            time_end = None

        # 目标表配置
        target_database_name = target_config.get("database_name")
        target_table_name = target_config.get("table_name")
        target_join_field = target_config.get("join_field", "id")
        target_time_field = target_config.get("time_field", "etl_date")
        filter_conditions = target_config.get("filter_conditions", [])
        stat_fields = target_config.get("stat_fields", {})

        if not all([target_database_name, target_table_name]):
            raise ValueError("目标表配置不完整：缺少数据库名或表名")

        # 解析统计字段配置
        if isinstance(stat_fields, str):
            try:
                stat_fields = json.loads(stat_fields) if stat_fields.strip() else {}
            except Exception:
                stat_fields = {}
        elif not isinstance(stat_fields, dict):
            stat_fields = {}

        # 获取统计字段映射和字段标签
        stat_field_mappings = {}
        try:
            if self.template_code:
                tpl = ReportTemplateService.get_template_by_code(self.template_code)
                if tpl:
                    default_params = tpl.get("default_params")
                    if isinstance(default_params, str):
                        try:
                            default_params = json.loads(default_params)
                        except Exception:
                            default_params = {}
                    if isinstance(default_params, dict):
                        stat_field_mappings = default_params.get(
                            "stat_field_mappings", {}
                        )
        except Exception:
            pass

        # 构建源表SQL（传递统计字段映射）
        source_union_sql, source_ids_sql, source_table_list = self._build_source_union_sql(
            source_tables, target_join_field, time_start, time_end, stat_field_mappings
        )

        # 根据源表是否启用去重，生成源表总数统计表达式
        use_distinct_for_source = False
        if isinstance(source_tables, dict):
            databases = source_tables.get("databases", [])
            database_name = source_tables.get("database_name")
            table_names = source_tables.get("table_names", [])
            tables_config = source_tables.get("tables", [])

            # 多数据库配置：检查每个表的 use_distinct
            for db_config in databases:
                tbls_config = db_config.get("tables", [])
                for tbl_config in tbls_config:
                    if tbl_config.get("use_distinct"):
                        use_distinct_for_source = True
                        break
                if use_distinct_for_source:
                    break

            # 单数据库多表配置（兼容旧格式）
            if not use_distinct_for_source and database_name and table_names:
                for tbl_name in table_names:
                    tbl_config = next(
                        (t for t in tables_config if t.get("table_name") == tbl_name),
                        {},
                    )
                    if tbl_config.get("use_distinct"):
                        use_distinct_for_source = True
                        break

        # 列表格式配置（兼容旧格式）
        elif isinstance(source_tables, list):
            for source_table in source_tables:
                if source_table.get("use_distinct"):
                    use_distinct_for_source = True
                    break

        if use_distinct_for_source:
            source_total_count_expr = f"COUNT(DISTINCT {target_join_field})"
        else:
            source_total_count_expr = "COUNT(*)"

        # 获取源表统计字段的输出字段列表（从所有表配置中收集）
        source_stat_output_fields = []
        if isinstance(source_tables, dict):
            databases = source_tables.get("databases", [])
            for db_config in databases:
                tbls_config = db_config.get("tables", [])
                # 从每个表配置中收集统计字段
                for tbl_config in tbls_config:
                    tbl_stat_fields = tbl_config.get("stat_fields", {})
                    if isinstance(tbl_stat_fields, str):
                        try:
                            tbl_stat_fields = json.loads(tbl_stat_fields) if tbl_stat_fields.strip() else {}
                        except Exception:
                            tbl_stat_fields = {}
                    elif not isinstance(tbl_stat_fields, dict):
                        tbl_stat_fields = {}
                    
                    # 如果表配置中没有，则使用数据库级别的配置（向后兼容）
                    if not tbl_stat_fields:
                        db_stat_fields = db_config.get("stat_fields", {})
                        if isinstance(db_stat_fields, str):
                            try:
                                db_stat_fields = json.loads(db_stat_fields) if db_stat_fields.strip() else {}
                            except Exception:
                                db_stat_fields = {}
                        elif not isinstance(db_stat_fields, dict):
                            db_stat_fields = {}
                        tbl_stat_fields = db_stat_fields
                    
                    if tbl_stat_fields:
                        _, output_fields = self._build_stat_fields_sql(
                            tbl_stat_fields, stat_field_mappings, prefix="source_"
                        )
                        source_stat_output_fields.extend(output_fields)
        
        # 构建源表显示字符串（多个源表用逗号分隔）
        source_tables_display = ", ".join(source_table_list) if source_table_list else ""
        target_table_display = f"{target_database_name}.{target_table_name}"

        # 构建目标表筛选条件
        where_conditions = []
        where_conditions.append(
            self._build_filter_conditions(filter_conditions)
        )
        target_time_condition = self._build_time_condition(
            time_start, time_end, target_time_field
        )
        if target_time_condition:
            where_conditions.append(target_time_condition)

        where_clause = "".join(where_conditions)

        # 构建目标表统计字段SQL
        stat_select_clause, stat_output_fields = self._build_stat_fields_sql(
            stat_fields, stat_field_mappings
        )

        # 获取字段标签
        field_labels = self._get_field_labels()

        # 默认字段标签
        default_labels = {
            "source_table": "源表",
            "target_table": "目标表",
            "source_total_count": "源表总数",
            "target_total_count": "目标表总数",
            "sync_rate": "同步率",
            "data_status": "数据状态",
        }
        for key, value in default_labels.items():
            if key not in field_labels:
                field_labels[key] = value

        # 为统计字段添加标签（目标表和源表）
        for output_field in stat_output_fields:
            if output_field not in field_labels:
                field_labels[output_field] = output_field
        for output_field in source_stat_output_fields:
            if output_field not in field_labels:
                field_labels[output_field] = output_field

        # 同步率阈值
        sync_rate_threshold = filter_rules.get("sync_rate_threshold", 80)

        # 构建统计字段的SELECT子句（用于最终输出）
        stat_fields_select = ""
        if stat_output_fields:
            stat_select_parts = []
            for output_field in stat_output_fields:
                # 跳过 target_total_count，因为它已经在主SELECT中
                if output_field == "target_total_count":
                    continue
                escaped_output_field = escape_column_name(output_field)
                label = field_labels.get(output_field, output_field)
                stat_select_parts.append(
                    f"COALESCE(target.{escaped_output_field}, 0) AS '{label}'"
                )
            if stat_select_parts:
                stat_fields_select = ",\n    " + ",\n    ".join(stat_select_parts)
        
        # 构建源表统计字段的SELECT子句（用于最终输出）
        source_stat_fields_select = ""
        if source_stat_output_fields:
            source_stat_select_parts = []
            for output_field in source_stat_output_fields:
                escaped_output_field = escape_column_name(output_field)
                label = field_labels.get(output_field, output_field)
                source_stat_select_parts.append(
                    f"COALESCE(src.{escaped_output_field}, 0) AS '{label}'"
                )
            if source_stat_select_parts:
                source_stat_fields_select = ",\n    " + ",\n    ".join(source_stat_select_parts)

        # 尝试使用SQL模板
        sql_template = self._get_sql_template()
        if sql_template:
            try:
                # 构建源表统计字段的SELECT（在source_union CTE中）
                source_stat_select_in_union = ""
                if source_stat_output_fields:
                    source_stat_parts = []
                    for output_field in source_stat_output_fields:
                        escaped_output_field = escape_column_name(output_field)
                        source_stat_parts.append(f"SUM({escaped_output_field}) AS {escaped_output_field}")
                    if source_stat_parts:
                        source_stat_select_in_union = ", " + ", ".join(source_stat_parts)
                
                render_ctx = {
                    "source_union_sql": source_union_sql,
                    "source_ids_sql": source_ids_sql,
                    "target_table_ref": f"`{target_database_name}`.`{target_table_name}`",
                    "join_field": target_join_field,
                    "source_total_count_expr": source_total_count_expr,
                    "where_clause": where_clause,
                    "stat_select_clause": stat_select_clause,
                    "stat_fields_select": stat_fields_select,
                    "source_stat_select_in_union": source_stat_select_in_union,
                    "source_stat_fields_select": source_stat_fields_select,
                    "stat_output_fields": stat_output_fields,
                    "source_stat_output_fields": source_stat_output_fields,
                    "sync_rate_threshold": sync_rate_threshold,
                    "source_tables_display": source_tables_display,
                    "target_table_display": target_table_display,
                    "source_table_label": field_labels.get("source_table", "源表"),
                    "target_table_label": field_labels.get("target_table", "目标表"),
                    "source_total_count_label": field_labels.get(
                        "source_total_count", "源表总数"
                    ),
                    "target_total_count_label": field_labels.get(
                        "target_total_count", "目标表总数"
                    ),
                    "sync_rate_label": field_labels.get("sync_rate", "同步率"),
                    "data_status_label": field_labels.get("data_status", "数据状态"),
                }
                return sql_template.format_map(SafeDict(render_ctx))
            except Exception as exc:
                get_logger().warning(
                    f"渲染SQL模板失败，使用默认SQL: {exc}", exc_info=True
                )

        # 默认SQL（如果没有模板或模板渲染失败）
        target_table_ref = f"`{target_database_name}`.`{target_table_name}`"
        escaped_target_total_count = escape_column_name('target_total_count')
        source_table_label = field_labels.get("source_table", "源表")
        target_table_label = field_labels.get("target_table", "目标表")
        
        # 构建源表统计字段的SELECT（在source_union CTE中）
        source_stat_select_in_union = ""
        if source_stat_output_fields:
            source_stat_parts = []
            for output_field in source_stat_output_fields:
                escaped_output_field = escape_column_name(output_field)
                source_stat_parts.append(f"SUM({escaped_output_field}) AS {escaped_output_field}")
            if source_stat_parts:
                source_stat_select_in_union = ", " + ", ".join(source_stat_parts)
        
        sql = f"""
WITH source_union AS (
    SELECT 
        {source_total_count_expr} AS source_total_count{source_stat_select_in_union}
    FROM (
        {source_union_sql}
    ) t
),
source_ids AS (
    SELECT DISTINCT {target_join_field}
    FROM (
        {source_ids_sql}
    ) t
),
target_stats AS (
    SELECT
        {stat_select_clause}
    FROM {target_table_ref} target
    WHERE EXISTS (
        SELECT 1
        FROM source_ids src
        WHERE src.{target_join_field} = target.{target_join_field}
    ){where_clause}
)
SELECT
    '{source_tables_display}' AS '{source_table_label}',
    '{target_table_display}' AS '{target_table_label}',
    src.source_total_count AS '{field_labels.get("source_total_count", "源表总数")}',
    COALESCE(target.{escaped_target_total_count}, 0) AS '{field_labels.get("target_total_count", "目标表总数")}',{source_stat_fields_select}{stat_fields_select}
    CONCAT(
        ROUND(
            CASE
                WHEN src.source_total_count > 0
                THEN (COALESCE(target.{escaped_target_total_count}, 0) * 100.0 / src.source_total_count)
                ELSE 0
            END,
            2
        ),
        '%'
    ) AS '{field_labels.get("sync_rate", "同步率")}',
    CASE
        WHEN src.source_total_count = 0 THEN '无原始数据'
        WHEN COALESCE(target.{escaped_target_total_count}, 0) = 0 THEN '未同步'
        WHEN (COALESCE(target.{escaped_target_total_count}, 0) * 100.0 / src.source_total_count) < {sync_rate_threshold} THEN '同步率过低'
        ELSE '正常'
    END AS '{field_labels.get("data_status", "数据状态")}'
FROM source_union src, target_stats target
"""
        return sql.strip()

    def collect(self) -> Dict[str, Any]:
        """收集数据"""
        sql_to_execute = self._build_sql()

        target_config = self.params.get("target_table_config", {})
        target_data_source_id = target_config.get("data_source_id")

        if not target_data_source_id:
            raise ValueError("目标表数据源ID不能为空")

        get_logger().info(f"执行通用数据质量监控SQL: {sql_to_execute[:200]}...")

        try:
            with self.get_data_source_cursor(target_data_source_id) as cursor:
                cursor.execute(sql_to_execute)
                result = cursor.fetchone()

                if not result:
                    result = {
                        "source_total_count": 0,
                        "target_total_count": 0,
                        "sync_rate": "0%",
                        "data_status": "无原始数据",
                    }

                # 获取字段标签映射（用于反向查找）
                field_labels = self._get_field_labels()

                # 处理显示字段
                display_fields_raw = self.params.get("display_fields", [])
                normalized_display_fields = self._normalize_and_sort_display_fields(
                    display_fields_raw, field_labels
                )
                if normalized_display_fields:
                    filtered_result = {}

                    def _get_result_value(field_name: str) -> Optional[Any]:
                        """根据字段名或标签在查询结果中查找对应的值"""
                        possible_keys = []
                        # 1. 直接使用字段名（兼容输出字段与列名一致的情况）
                        possible_keys.append(field_name)

                        # 2. 使用字段标签（大多数列在SQL中使用标签作为别名）
                        field_label = field_labels.get(field_name)
                        if field_label:
                            possible_keys.append(field_label)

                        # 3. 如果 display_fields 中本身就是标签，尝试通过标签反查字段名
                        for key, label in field_labels.items():
                            if label == field_name:
                                possible_keys.append(label)
                                possible_keys.append(key)

                        for key in possible_keys:
                            if key in result:
                                return result[key]
                        return None

                    for field in normalized_display_fields:
                        value = _get_result_value(field)
                        if value is None:
                            continue
                        filtered_result[field] = value

                    if filtered_result:
                        result = filtered_result

                return {
                    "success": True,
                    "data": [result] if result else [],
                }
        except Exception as exc:
            get_logger().error(
                f"执行通用数据质量监控查询失败: {str(exc)}", exc_info=True
            )
            raise ValueError(f"执行查询失败: {str(exc)}")

    def generate_summary(self, data: Dict[str, Any]) -> Dict[str, Any]:
        """生成数据摘要"""
        summary = data.get("summary", {})
        return {
            "source_total_count": summary.get("source_total_count", 0),
            "target_total_count": summary.get("target_total_count", 0),
            "sync_rate": summary.get("sync_rate", "0%"),
            "data_status": summary.get("data_status", "未知"),
        }

